package io.gitee.tziye.core.bloom;

import com.google.common.base.Charsets;
import com.google.common.hash.Hashing;
import com.google.common.primitives.Longs;
import io.gitee.tziye.common.RenaiUtil;
import io.vavr.Tuple2;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class GuavaBloomFilter {

    private static final int LONG_ADDRESSABLE_BITS = 6;
    private static final int bitSize = 258816;
    private static final int numHashFunctions = 10;

    public static <T> List<Tuple2<Integer, Long>> calculate(T value) {
        byte[] bytes = hash(value);
        long hash1 = lowerEight(bytes);
        long hash2 = upperEight(bytes);

        List<Tuple2<Integer, Long>> list = new ArrayList<>();
        Set<Integer> set = new HashSet<>();
        long combinedHash = hash1;
        for (int i = 0; i < numHashFunctions; i++) {
            long bitIndex = (combinedHash & Long.MAX_VALUE) % bitSize;
            int longIndex = (int) (bitIndex >>> LONG_ADDRESSABLE_BITS);
            long mask = 1L << bitIndex;
            if (!set.contains(longIndex)) {
                list.add(new Tuple2<>(longIndex, mask));
                set.add(longIndex);
            }
            combinedHash += hash2;
        }
        return list;
    }

    public static <T> byte[] hash(T value) {
        return Hashing.murmur3_128().hashObject(value, (v, into) -> into
                .putString(RenaiUtil.string(v), Charsets.UTF_8)).asBytes();
    }

    private static long lowerEight(byte[] bytes) {
        return Longs.fromBytes(
                bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
    }

    private static long upperEight(byte[] bytes) {
        return Longs.fromBytes(
                bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);
    }

}
